package com.binlog.core.listener;

import com.alibaba.fastjson.JSON;
import com.binlog.config.BinLogDbProperty;
import com.binlog.config.BinLogDbPropertys;
import com.binlog.core.handler.CommonBackups;
import com.binlog.core.handler.HandlerFactory;
import com.binlog.module.BinLogItem;
import com.binlog.module.DatebaseInfo;
import com.binlog.module.TableColum;
import com.binlog.util.MysqlUtils;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

import static com.github.shyiko.mysql.binlog.event.EventType.*;

/**
 * @author cyj
 * @Description 通用项目监听
 * @create 2022-05-13 上午 11:47
 */
@Slf4j
public class CommonEventListener extends CommonBackups implements BinaryLogClient.EventListener {

    /**
     * 这个监听的表结构
     */
    private Map<String, Map<String, TableColum>> schemaTable;

    private BinLogDbProperty binLogDbProperty;

    private DatebaseInfo datebaseInfo;

    /**
     * 这个监听这个库tableId和table关系
     */
    private Map<Long, String> tableIdMapping = new ConcurrentHashMap();

    public CommonEventListener(Map<String, Map<String, TableColum>> schemaTable, BinLogDbProperty binLogDbProperty, DatebaseInfo datebaseInfo) {
        this.schemaTable = schemaTable;
        this.binLogDbProperty = binLogDbProperty;
        this.datebaseInfo = datebaseInfo;
    }

    @Override
    public void onEvent(Event event) {

        EventType eventType = event.getHeader().getEventType();

        if (eventType == EventType.TABLE_MAP) {
            TableMapEventData tableData = event.getData();
            if (binLogDbProperty.getSchema().equals(tableData.getDatabase()) && schemaTable.containsKey(tableData.getTable())) {
                tableIdMapping.putIfAbsent(tableData.getTableId(), tableData.getTable());
                log.info("刷新tableMap, 当前tableIdMapping：{}", JSON.toJSONString(event));
            }
        }

        EventData data = event.getData();

        // 检查EventData的类型是否为RotateEventData
        if (data instanceof RotateEventData) {
            RotateEventData rotateEventData = (RotateEventData) data;
            String binlogFilename = rotateEventData.getBinlogFilename();
            log.info("当前 Binlog Filename: " + binlogFilename);
        }

        // 只处理添加删除更新三种操作
        this.handle(event, eventType);

    }

    /**
     * 业务处理模板方法
     *
     * @param event
     * @param eventType
     */
    private void handle(Event event, EventType eventType) {
        EventHeaderV4 header = event.getHeader();
        long position = header.getPosition();
        if (log.isDebugEnabled()) {
            log.info("binlog accept! position:{}", header.getPosition());
        }
        if (isWrite(eventType)) {
            this.writeHandle(event, eventType, position);
        }
        if (isUpdate(eventType)) {
            this.updateHandle(event, eventType, position);
        }
        if (isDelete(eventType)) {
            this.deleteHandle(event, eventType, position);
        }
    }

    private void writeHandle(Event event, EventType eventType, long position) {

        WriteRowsEventData data = event.getData();
        long tableId = data.getTableId();
        if (!checkHandle(tableId)) {
            return;
        }
        log.info("监听到binlog add信息：{}", JSON.toJSONString(event));
        String table = tableIdMapping.get(tableId);

        for (Serializable[] row : data.getRows()) {
            BinLogItem item = BinLogItem.itemFromInsertOrDeleted(row, schemaTable.get(table), eventType, position);
            try {
                log.info("处理binlog add信息：{}", JSON.toJSONString(item));
                super.insertHandle(item);
                HandlerFactory.getInstance(table, binLogDbProperty.getSchema()).insertHandle(item);
            } catch (Exception e) {
                log.error("处理binlog add信息失败：{}, e:{}", JSON.toJSONString(item), e.getMessage(), e);
            }
        }

    }

    private void updateHandle(Event event, EventType eventType, long position) {

        UpdateRowsEventData data = event.getData();
        long tableId = data.getTableId();
        if (!checkHandle(tableId)) {
            return;
        }
        log.info("监听到binlog update信息：{}", JSON.toJSONString(event));

        String table = tableIdMapping.get(tableId);
        for (Map.Entry<Serializable[], Serializable[]> row : data.getRows()) {
            BinLogItem item = BinLogItem.itemFromUpdate(row, schemaTable.get(table), eventType, position);
            try {
                log.info("处理binlog update信息：{}", JSON.toJSONString(item));
                super.updateHandle(item);
                HandlerFactory.getInstance(table, binLogDbProperty.getSchema()).updateHandle(item);
            } catch (Exception e) {
                log.error("处理binlog update信息失败：{}, e:{}", JSON.toJSONString(item), e.getMessage(), e);
            }
        }
    }

    public void deleteHandle(Event event, EventType eventType, long position) {
        DeleteRowsEventData data = event.getData();
        long tableId = data.getTableId();

        if (!checkHandle(tableId)) {
            return;
        }

        log.info("监听到binlog delete信息：{}", JSON.toJSONString(event));
        String table = tableIdMapping.get(tableId);
        for (Serializable[] row : data.getRows()) {
            BinLogItem item = BinLogItem.itemFromInsertOrDeleted(row, schemaTable.get(table), eventType, position);
            try {
                log.info("处理binlog delete信息：{}", JSON.toJSONString(item));
                super.deleteHandle(item);
                HandlerFactory.getInstance(table, binLogDbProperty.getSchema()).deleteHandle(item);
            } catch (Exception e) {
                log.error("处理binlog delete信息失败：{}, e:{}", JSON.toJSONString(item), e.getMessage(), e);
            }
        }

    }


    /**
     * 校验
     */
    private boolean checkHandle(long tableId) {
        String table = tableIdMapping.get(tableId);
        return schemaTable.containsKey(table);
    }

    @Override
    public DatebaseInfo getDatebaseInfo() {
        return this.datebaseInfo;
    }

    @Override
    public BinLogDbProperty getBinLogDbProperty() {
        return this.binLogDbProperty;
    }
}
